package com.hyw.trace.configuration.meta.redis.queue;

import java.lang.reflect.Constructor;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.function.Consumer;
import java.util.function.Function;

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.Assert;

/**
 * 一个基于Redis的队列构造器，目前支持二种：
 * <li>SimpleRedisQueue
 * <p>
 * 简单的队列，符合FIFO
 * <li>IdempotentRedisQueue
 * <p>
 * 冪等队列,依赖于key值所对应的Value做去重判定，如果新增的key所对应的value与当前维护的value相同，则认定为重复
 * 
 * @author Hongyu
 */
public class RedisQueueBuilder {

  private String topic;

  private RedisTemplate<String, String> redisTemplate;

  private Class<? extends AbstractCacheQueue> type;

  private Consumer<String> consumer;

  private Function<String, Entry<String, String>> keyExtractor;

  protected RedisQueueBuilder() {
    super();
  }

  public static RedisQueueBuilder create(String topic, RedisTemplate<String, String> redisTemplate) {
    RedisQueueBuilder b = new RedisQueueBuilder();
    b.topic = topic;
    b.redisTemplate = redisTemplate;
    b.type = SimpleRedisQueue.class;
    return b;
  }

  /**
   * 如果入参值{@link IdempotentRedisQueue}时，正常应该设置{@link #setKeyExtractor(Function)}.如果没有设置，在创建对象时，会抛出异常
   * 
   * @param clz
   * @return
   */
  public final RedisQueueBuilder setQueueType(QueueType type) {
    Objects.requireNonNull(type);
    this.type = type.getClz();
    return this;
  }

  /**
   * 只有当{@link #setQueueType(Class)} 入参为{@link IdempotentRedisQueue}时有效
   * 
   * @param keyExtractor
   * @return
   */
  public final RedisQueueBuilder setKeyExtractor(Function<String, Entry<String, String>> keyExtractor) {
    this.keyExtractor = keyExtractor;
    return this;
  }

  public final RedisQueueBuilder setConsumer(Consumer<String> consumer) {
    this.consumer = consumer;
    return this;
  }

  public final AbstractCacheQueue build() {
    AbstractCacheQueue instance = null;
    try {
      if (type == IdempotentRedisQueue.class) {
        Assert.isTrue(Objects.nonNull(keyExtractor), "IdempotentRedisQueue 必须设置 keyExtractor");
        Constructor<?> ct = type.getDeclaredConstructor(Function.class, String.class, RedisTemplate.class,
            Consumer.class);
        instance = (AbstractCacheQueue) ct.newInstance(this.keyExtractor, this.topic, this.redisTemplate,
            this.consumer);
      } else if (type == SimpleRedisQueue.class) {
        Constructor<?> ct = type.getDeclaredConstructor(String.class, RedisTemplate.class, Consumer.class);
        instance = (AbstractCacheQueue) ct.newInstance(this.topic, this.redisTemplate, this.consumer);
      } else {
        throw new UnsupportedOperationException();
      }

    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }

    if (Objects.nonNull(consumer)) {
      AbstractCacheQueue queue = instance;
      new Thread(queue::run).start();
    }
    return instance;
  }

  public enum QueueType {

    Simple(SimpleRedisQueue.class), Idempotent(IdempotentRedisQueue.class);

    Class<? extends AbstractCacheQueue> clz;

    private QueueType(Class<? extends AbstractCacheQueue> clz) {
      this.clz = clz;
    }

    public Class<? extends AbstractCacheQueue> getClz() {
      return clz;
    }

  }
}
